package com.JadePenG.Storm;

import com.alibaba.fastjson.JSON;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 接收数据, 处理数据
 */
public class WordCountBolt extends BaseBasicBolt {

    //问题: 放map不合适
    private HashMap<String, Integer> map = new HashMap<>();

    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        //1. 获取上游数据
        String keywords = tuple.getStringByField("keywords");
        //2. 统计次数
        if (map.get(keywords) != null && map.get(keywords) > 0) {
            map.put(keywords, map.get(keywords) + 1);
        } else {
            map.put(keywords, 1);
        }
        //3. 发送数据到下游，这个位置不能给map直接发走，下游是kafkaBolt只负责发送数据
        //在这个位置就要是一个最终的数据  topic:keywords "[{'topKeywords':'','score':99},{}]"
        //[] -->list {}-->对应对象, 有没有pojo 有:直接封装 没有: Map   与json转换
        List<Map<String, Object>> resultList = new ArrayList<>();

        //将map的结果放到resultList(发送走的对象)中,然后将resultList变成字符串发送走
        //1. 将map中的数据封装到list中
        for (String key : map.keySet()) {
            //key: 搜索的内容 keywords score次数
            Integer score = map.get(key);
            Map<String, Object> resultMap = new HashMap<>();
            resultMap.put("topKeywords", key);
            resultMap.put("score", score);
            resultList.add(resultMap);
        }

        //2. 将list变成json字符串发走
        String jsonString = JSON.toJSONString(resultList);
        System.out.println(jsonString);
        basicOutputCollector.emit(new Values(jsonString));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //名字是固定的 message
        outputFieldsDeclarer.declare(new Fields("message"));
    }
}
